package org.jsets.fastboot.eventbus;

import java.util.concurrent.DelayQueue;

import org.jsets.fastboot.concurrent.AsyncActuator;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class RetryQueue {

	private final DelayQueue<RetryTask> queue = new DelayQueue<RetryTask>();

	private static class SingleHolder {
		private static final RetryQueue INSTANCE = new RetryQueue();
	}

	public static RetryQueue getInstance() {
		return SingleHolder.INSTANCE;
	}

	private RetryQueue() {
		this.startMonitoring();
	}

	private void startMonitoring() {
		new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						RetryTask task = queue.take();
						doRetry(task);
						task = null;
					} catch (Exception e) {
						// 忽略... ...
					}
				}
			}
		}, "RetryQueueMonitor-Thread").start();
	}

	private void doRetry(RetryTask task) {
		AsyncActuator.execute(new Runnable() {
			@Override
			public void run() {
				Object event = task.getEvent();
				String subscriberKey = task.getSubscriberKey();
				Subscriber subscriber = SubscriberRegistry.getSubscriber(subscriberKey);
				DefaultEventBus bus = subscriber.getBus();
				log.info("第" + task.getStep() + "次重试，事件[" + event.toString() + "]，监听器["+ subscriber.getTarget().toString() + "]");
				try {
					subscriber.invokeSubscriberMethod(event);
					log.info("事件处理成功，事件["+event.toString()+"]，监听器["+subscriber.getTarget().toString()+"]");
				} catch (Exception e) {
					boolean necessary = subscriber.retryNecessary(e);
					if (necessary && task.getStep() < subscriber.getMaxAttempts()) {
						long exponent = task.getDelay() * subscriber.getMultiplier();
						int step = task.getStep() + 1;
						RetryTask newTask = new RetryTask(event, subscriberKey, exponent, step);
						queue.offer(newTask);
					} else {
						bus.handleSubscriberException(e.getCause(), new SubscriberExceptionContext(bus, event, subscriber.getTarget(), subscriber.getMethod()));
					}
				}
			}

		});

	}

	public void put(RetryTask event) {
		if (null != event) {
			this.queue.put(event);
		}
	}
}
